package com.taobao.udp.udf;

import java.util.ArrayList;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import ws.Wsjni;

public class GenericUDTFWJS extends GenericUDTF {

	private Wsjni ws;
	private boolean isInit = false;

	@Override
	public StructObjectInspector initialize(ObjectInspector[] argOIs)
			throws UDFArgumentException {
		// TODO Auto-generated method stub
		if (argOIs.length != 1) {
			throw new UDFArgumentException(
					"GenericUDTFWJS() takes only one argument");
		}
		if (argOIs[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
			throw new UDFArgumentException(
					"GenericUDTFWJS() takes only primitive argument");
		}
		ArrayList<String> fieldNames = new ArrayList<String>();
		ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
		fieldNames.add("col");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		return ObjectInspectorFactory.getStandardStructObjectInspector(
				fieldNames, fieldOIs);
	}

	private void initWs() throws Exception {
	    isInit = true;
		ws = new Wsjni();
		if (ws.ws_init("./ws/taobao.conf","TAOBAO_CHN") != 0) {
			throw new HiveException("init failed");
		}
	}

	@Override
	public void process(Object[] args) throws HiveException {
		// TODO Auto-generated method stub
		if(!isInit){
			try {
				initWs();
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		String str = args[0].toString();
		if(!StringUtils.isBlank(str)){
			Object[] ret = ws.ws_segment_arr(str, 2);
			if(ret != null){
				for(Object o : ret){
					String[] tmp = {String.valueOf(o)};
					forward(tmp);
					tmp = null;
				}
				ret = null;
				return;
			}
		}
		forward(new String[]{""});
	}

	@Override
	public void close() throws HiveException {
		// TODO Auto-generated method stub
		if(ws != null){
			ws.ws_destroy();
		}
	}
	
}
